package app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.UserCommentBean;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DWSInteractionComment extends BaseApp {
    public static void main(String[] args) {
        new DWSInteractionComment().start(10021,4,"omm","dwd_interaction_comment");
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(s)) {
                            JSONObject jsonObject = JSON.parseObject(s);
                            collector.collect(jsonObject);
                        }
                    }
                }
        );


        SingleOutputStreamOperator<UserCommentBean> beanDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts") * 1000;
                                    }
                                }
                        )
        ).keyBy(jsonObject -> jsonObject.getString("user_id")).map(
                new MapFunction<JSONObject, UserCommentBean>() {
                    @Override
                    public UserCommentBean map(JSONObject jsonObject) throws Exception {
                        String course_id = jsonObject.getString("course_id");
                        String review_stars = jsonObject.getString("review_stars");
                        double stars = Double.parseDouble(review_stars);
                        String comment_txt = jsonObject.getString("comment_txt");

                        return new UserCommentBean("","","",course_id, "", stars, 1L, comment_txt.equals("评论 123123") ? 1L : 0L, 0d);
                    }
                }
        ).//print();
                keyBy(UserCommentBean::getCourse_id).window(
                TumblingEventTimeWindows.of(Time.seconds(5))
        ).reduce(
                new ReduceFunction<UserCommentBean>() {
                    @Override
                    public UserCommentBean reduce(UserCommentBean v1, UserCommentBean v2) throws Exception {
                        v1.setStars(v1.getStars() + v2.getStars());
                        v1.setUser_count(v1.getUser_count() + v2.getUser_count());
                        v1.setComment_count(v1.getComment_count() + v2.getComment_count());
                        return v1;
                    }
                },
                new WindowFunction<UserCommentBean, UserCommentBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<UserCommentBean> iterable, Collector<UserCommentBean> collector) throws Exception {
                        UserCommentBean bean = iterable.iterator().next();
                        bean.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        collector.collect(bean);
                    }
                }
        ).map(
                new MapFunction<UserCommentBean, UserCommentBean>() {
                    @Override
                    public UserCommentBean map(UserCommentBean userCommentBean) throws Exception {
                        userCommentBean.setAvg_starts(userCommentBean.getStars() / userCommentBean.getUser_count());
                        return userCommentBean;
                    }
                }
        );
        SingleOutputStreamOperator<UserCommentBean> sinkDS = AsyncDataStream.unorderedWait(
                beanDS,
                new DimAsyncFunction<UserCommentBean>() {
                    @Override
                    protected void addDim(UserCommentBean obj, JSONObject dimJsonObj) {
                        obj.setCourse_name(dimJsonObj.getString("course_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getRowKey(UserCommentBean obj) {
                        return obj.getCourse_id();
                    }
                }, 60, TimeUnit.SECONDS
        );
        sinkDS.print();
        sinkDS.map(new BeanToJsonStrMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_interaction_course_comment_window"));


    }
}
